package com.pccc.batch.config.remotepartition;

import java.util.function.Function;

import javax.sql.DataSource;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.partition.support.RemoteStepExecutionAggregator;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.client.OkHttp3ClientHttpRequestFactory;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.http.converter.SerializingHttpMessageConverter;
import org.springframework.integration.http.dsl.Http;
import org.springframework.integration.http.support.DefaultHttpHeaderMapper;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.messaging.PollableChannel;
import org.springframework.transaction.PlatformTransactionManager;

import com.pccc.batch.config.remotepartition.handler.PollerRemotePartitioningManagerStepBuilderFactory;
import com.pccc.batch.config.remotepartition.listener.MyJobListener;
import com.pccc.batch.config.remotepartition.pontioner.FilePartitioner;

@Configuration
@EnableBatchIntegration
@EnableBatchProcessing
public class MasterConfiguration {

	
	private final JobBuilderFactory jobBuilderFactory;
	
	private final JobExplorer jobExplorer;
	
	private final JobRepository jobRepository;
	
	private final PlatformTransactionManager transactionManager;
	
	private final DataSource dataSource;
	
	public MasterConfiguration(JobBuilderFactory jobBuilderFactory,JobExplorer jobExplorer,JobRepository jobRepository,
			PlatformTransactionManager transactionManager,DataSource dataSource) {
		this.jobBuilderFactory = jobBuilderFactory;
		this.jobExplorer = jobExplorer;
		this.dataSource = dataSource;
		this.transactionManager = transactionManager;
		this.jobRepository = jobRepository;
	}
	
	
	//----------------------------------------------------------------
    //  partitionJob 配置
    //----------------------------------------------------------------
    @Bean
    public Job partitionJob(JobBuilderFactory jobs,
                            @Qualifier("masterStep") Step masterStep,
                            @Qualifier("slaveStep") Step slaveStep) {
        return jobs.get("partitionJob")
                .start(masterStep)
                .incrementer(new RunIdIncrementer())
                .listener(new MyJobListener())
                .build();
    }
    
    
    
    @Bean(name = "masterStep")
    public Step masterStep(@Qualifier("slaveStep") Step slaveStep) {
    	return managerStepBuilderFactory().get("masterStep")
						    	.partitioner(slaveStep.getName(), new FilePartitioner())
						        .step(slaveStep)
						        .gridSize(1)
						        .aggregator(stepExecutionAggregator())
						        .outputChannel(requests())
						        .inputChannel(reply())
						        .build();
    }
    
    @Bean
    public PollerRemotePartitioningManagerStepBuilderFactory managerStepBuilderFactory() {
    	return new PollerRemotePartitioningManagerStepBuilderFactory(jobExplorer, jobRepository, transactionManager, dataSource);
    }
    
    
    @Bean
    public RemoteStepExecutionAggregator stepExecutionAggregator(){
        return new RemoteStepExecutionAggregator(jobExplorer);
    }
	
    @Bean
	public PollableChannel reply() {
		return new QueueChannel(10);
	}
    
    
    @Bean(name = PollerMetadata.DEFAULT_POLLER)
    public PollerMetadata poller() {
       return Pollers.fixedRate(500).get();
    }
    
    
    @Bean
    public IntegrationFlow httpIn() {
		return IntegrationFlows
				.from(Http.inboundGateway("/reply")
						  .requestChannel(reply())
						  .requestMapping(m -> {
								m.consumes("application/x-java-serialized-object");
								m.methods(HttpMethod.POST);
							})
						  .headerMapper(httpMapper())
						  .headerFunction("correlationId", new Function<HttpEntity<String>, String>() {
							@Override
							public String apply(HttpEntity<String> t) {
								return t.getHeaders().getFirst("correlationid");
							}
						})
						  .headerFunction("sequenceNumber", new Function<HttpEntity<String>, Integer>() {
								@Override
								public Integer apply(HttpEntity<String> t) {
									return Integer.parseInt(t.getHeaders().getFirst("sequencenumber"));
								}
							})
						  .headerFunction("sequenceSize", new Function<HttpEntity<String>, Integer>() {
								@Override
								public Integer apply(HttpEntity<String> t) {
									return Integer.parseInt(t.getHeaders().getFirst("sequencesize"));
								}
							})
						  .messageConverters(new SerializingHttpMessageConverter())
						  )
				.log()
				.get();
    }
    
    
    @Bean
	public DirectChannel requests() {
		return new DirectChannel();
	}
    
    @Bean
    public IntegrationFlow httpOut() {
        return IntegrationFlows.from(requests())
                .handle(Http.outboundGateway("http://localhost:8088/store")
                        .charset("UTF-8")
                        .httpMethod(HttpMethod.POST)
                        .messageConverters(new SerializingHttpMessageConverter())
                        .requestFactory(new OkHttp3ClientHttpRequestFactory())
                        .headerMapper(httpMapper())
                        )
                .log()
                .get();
    }
    
    @Bean
    public DefaultHttpHeaderMapper httpMapper() {
    	DefaultHttpHeaderMapper httpMapper = new DefaultHttpHeaderMapper();
    	httpMapper.setInboundHeaderNames("correlationId","replyChannel","sequenceNumber","sequenceSize","timestamp","id");
    	httpMapper.setOutboundHeaderNames("correlationId","replyChannel","sequenceNumber","sequenceSize","timestamp","id");
    	return httpMapper;
    }
    
}
